package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.constant.MqttDestinationConst;
import cn.tedu.charging.common.pojo.message.ChargingCheckResultMessage;
import cn.tedu.charging.common.pojo.message.ChargingProgressMsg;
import cn.tedu.charging.order.service.ConsumerService;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 给当前底层连接客户端MqttClient
 * 提供一个回调监听
 */
@Component
@Slf4j
public class MqttClientCallback implements MqttCallbackExtended {
    @Autowired
    private ConsumerService consumerService;
    @Autowired
    private RedisTemplate redisTemplate;
    /**
     * 回调时机: 客户端mqttClient成功建立与emqx服务端连接关系时调用
     * boolean reconnect: 是否是重连
     * String serverURI: emqx地址 tcp://192.168.8.100:1883
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if (reconnect){
            log.info("重新连接成功,broker地址:{}",serverURI);
        }else{
            log.info("首次连接成功,broker地址:{}",serverURI);
        }
    }

    /**
     * 回调时机: mqttClient连接断开时调用(被踢掉 或者网络波动)
     * Throwable cause: 断开的异常原因
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.info("连接断开",cause);
    }

    /**
     * 回调时机:mqttClient发送publish消息的时候,如果成功了,收到emqx反馈ack=true就会调用
     * 这个方法,但是只有在QOS>0的时候才有效 因为  QOS=0 客户端不关心成功失败,不接收反馈
     * IMqttDeliveryToken token: 消息投递发送所有包装数据封装
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        //isComplete()可以反馈发送消息动作是否完成 ack大概率会是true
        //如果消息发送失败的,怎么靠这个方法获取失败结果
        boolean ack=token.isComplete();
        //消息发送失败,返回exception 是空的
        MqttException exception = token.getException();
        if (ack&&exception==null){
            log.info("消息发送结束 且成功");
        }else if(ack&&exception!=null){
            log.info("消息发送结束 但失败");
        }else if(!ack){
            log.info("消息未执行发送");
        }
    }

    /**
     * 回调时机: 首先当前回调对象所属的客户端连接 必须订阅主题,否则messageArrived永远不调用.
     * 如果主题中出现消息,每一条消息都会在客户端回调对象调用一次这个方法
     * String topic: 这条消息的来源主题 不同主题可能处理业务逻辑不一样的
     * MqttMessage message: 消息传递对象 对方发送时是什么 这里接收就是什么需要将payLoad做反序列化
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("消息来源topic:{}",topic);
        if (topic!=null&&topic.equals(MqttDestinationConst.GUN_CHECK_RESULT_TOPIC)){
            log.debug("当前消息消费逻辑是 处理设备自检结果");
            //反序列化 解析消息对象 message获取body 变成json 变成ChargingCheckResultMessage
            byte[] payload = message.getPayload();
            String json=new String(payload);
            ChargingCheckResultMessage msg = JSON.parseObject(json, ChargingCheckResultMessage.class);
            //1.抢锁 保证lock的key值和延迟消息抢锁的key值相同结构
            String orderLock="charging:order:consume:"+msg.getOrderNo()+".lock";
            Boolean haveLock = redisTemplate.opsForValue().setIfAbsent(orderLock, "", 5, TimeUnit.SECONDS);
            if (haveLock){
                try{
                    consumerService.handleCheckResult(msg);
                }catch ( Exception e ){
                    log.error("设备自检消费失败",e);
                }finally{
                    //手动释放锁
                    redisTemplate.delete(orderLock);
                }
            }
        }else if(topic!=null&& topic.equals(MqttDestinationConst.CHARGING_PROGRESS_TOPIC)){
            //调用业务层 处理充电进度同步的信息
            byte[] payload = message.getPayload();
            String json=new String(payload);
            ChargingProgressMsg msg = JSON.parseObject(json, ChargingProgressMsg.class);
            //业务接口实现处理
            consumerService.handleChargingProgress(msg);
        }else{
            log.error("我没监听这个主题,闹鬼了");
        }
    }
}
